#include "stdafx.h"
#include "OracleAdapter.h"
#include "MessageTypes.h"
#include "XMLHelper.h"
#include "Metadata.h"
#include "Queries.h"
#include "ConnectionString.h"
#include "COMTools.h"
#include "ComponentRegister.h"
#include "MetaXmlHandler.h"
#include <sstream>
#include <iostream>
#include <comdef.h>
//#include "ORACLEADAPTER.h"

#import "C:\windows\system32\msado15.dll" rename_namespace("MSXML")

const char* MESSAGE_TEMPLATE = "<?xml version=\"1.0\" encoding=\"iso-8859-1\"?>"
                               "<!DOCTYPE message ["
                               "<!ELEMENT message (header, body)>"
                               "<!ELEMENT header (id?, date?, refid?, sender, status?, subject)>"
                               "<!ELEMENT id      (#PCDATA)>"
                               "<!ELEMENT date    (#PCDATA)>"
                               "<!ELEMENT sender  (#PCDATA)>"
                               "<!ELEMENT status  (#PCDATA)>"
                               "<!ELEMENT subject (#PCDATA)>"
                               "<!ELEMENT body    (#PCDATA)>"
                               "]>"
                               "<message>"
                               "<header>"
                               "<sender>%s</sender><subject>%s</subject>"
                               "</header>"
                               //"<body><![CDATA[%s]]></body>"
                               "<body>%s</body>"
                               "</message>";


SQLServerAdapter::SQLServerAdapter()
: DatabaseAdapter(UNKNOWN)
, m_pszTemp(NULL)
{

}

SQLServerAdapter::SQLServerAdapter(E_SUBJECT subj)
: DatabaseAdapter(subj)
, m_pszTemp(NULL)
{
  initConnection();
}

SQLServerAdapter::~SQLServerAdapter()
{
  try { cn->Close(); } catch (...) { }
  if (m_pszTemp) delete[] m_pszTemp;
  
  ComponentRegister::getInstance()->unregisterComponent(this->szSender);
}

void SQLServerAdapter::initConnection()
{
  HRESULT hr;
  bstr_t cns;
    
  ::CoInitialize(NULL);

  hr = cn.CreateInstance(__uuidof(ADODB::Connection), NULL, CLSCTX_INPROC_SERVER);  
  if (S_OK != hr) throw("CreateInstance failed");
  cns = bstr_t(CONNECTSTRING);

  while (!cnOpen) {
    try {
      hr = cn->Open(cns, L"", L"", -1);  
      if (S_OK != hr) throw("Database not opened properly");
      cnOpen = true;
    } catch (_com_error& e) {
      TCHAR buffer[0x1000] = { 0 };        
      sprintf(buffer, "%s [%ld]\n", e.ErrorMessage(), e.Error());  
      OutputDebugString(buffer);
      try { cn->Close(); } catch (...) { }
      cnOpen = false;
    }
    if (!cnOpen) ::Sleep(1000);
  }
}

void SQLServerAdapter::onMessage(infobus::types::Message * m)
{
  if (m) {
    std::stringstream ss;
    m->operator <<(ss);

    char * szXml = new char[strlen(ss.str().c_str()) +1];
    memset(szXml, 0, strlen(ss.str().c_str()) +1);
    strncpy(szXml, ss.str().c_str(), strlen(ss.str().c_str()));  

    DatabaseAdapter::getListener()->onMessage(szXml);
    delete[] szXml;
  }
}


void SQLServerAdapter::run()
{
  HRESULT hr = S_FALSE;
  while (!Thread::isTerminated() && waitForLock()) {
    try {
      bool bActive = true;
      if (m_bAliveBeats) {
        ComponentRegister::getInstance()->activity(this->getSender());
        if (ComponentRegister::getInstance()->isRedundant(this->getSender())) {
          bActive = ComponentRegister::getInstance()->isMaster(this->getSender());
          //if (!bActive) {
          //  bActive = !ComponentRegister::getInstance()->isMasterActive(this->getSender(), this->getSubjectText());
          //}
        }
      }
           
      infobus::types::Message * pMsg = NULL;

      std::stringstream ss;
      ss << "Looking for messages on channel["<< this->getSubjectText() <<"] ..."
         << std::endl;          
      OutputDebugString(ss.str().c_str());
      
      _bstr_t sql = _bstr_t(" SELECT TOP 1 *, len(convert(varchar(max),message)) * 2 AS SIZE FROM MESSAGE_QUEUE WHERE SUBJECT LIKE '"+ _bstr_t(SubjectText(subject)) +"' AND STATUS = '"+ _bstr_t(PENDING) +"' ORDER BY CREATED_TIME ");

      ADODB::_RecordsetPtr rs;  
      rs.CreateInstance (__uuidof(ADODB::Recordset));  
      rs->Open(sql, cn.GetInterfacePtr(), ADODB::adOpenDynamic, ADODB::adLockPessimistic, ADODB::adCmdText);  

      if (!rs->ADOEOF) hr = rs->MoveFirst();
      if (S_OK == hr && !rs->ADOEOF) {          
        pMsg = new infobus::types::Message(
          rs->Fields->GetItem(L"MESSAGE_ID")->GetValue(), 
          (E_STATUS)(int)rs->Fields->GetItem(L"STATUS")->GetValue(), 
          (E_SUBJECT)SubjectEnum(_bstr_t(rs->Fields->GetItem(L"SUBJECT")->GetValue())), 
          _bstr_t(rs->Fields->GetItem(L"SENDER")->GetValue()), 
          _bstr_t(rs->Fields->GetItem(L"MESSAGE")->GetValue())
        );
        
        pMsg->setTimestamp(_bstr_t(rs->Fields->GetItem(L"CREATED_TIME")->GetValue()));

        long chunkSize = rs->Fields->GetItem(L"SIZE")->GetValue();

        _variant_t val;
        BSTR bdata, xml = _bstr_t("");

        VariantInit(&val);
        val = rs->Fields->GetItem(L"MESSAGE")->GetChunk(chunkSize +2);
        COMTools::UnpackVariant(val, &bdata);
        VariantClear(&val);

        VARIANT_BOOL b;
        IXMLDOMDocumentPtr pXMLDOMDoc; 
        IXMLDOMNodePtr pXMLDOMNode = NULL;
        IXMLDOMNodeListPtr pXMLDOMNodeList = NULL;
        pXMLDOMDoc.CreateInstance(__uuidof(DOMDocument));
        pXMLDOMDoc->put_async(false);
        pXMLDOMDoc->put_resolveExternals(false);
        hr = pXMLDOMDoc->loadXML(bdata, &b);
        
        ::SysFreeString(bdata);
        
        if (S_OK == hr) hr = pXMLDOMDoc->selectNodes(_bstr_t("//body/*"), &pXMLDOMNodeList);
        if (S_OK == hr) hr = pXMLDOMNodeList->get_item(0, &pXMLDOMNode);
        if (S_OK == hr) hr = pXMLDOMNode->get_xml(&xml);        
        if (S_OK == hr) 
          pMsg->setMessage(_bstr_t(xml));
     
        if (bActive) {
          this->onMessage(pMsg);
          rs->Fields->Item[L"STATUS"]->Value = _variant_t((int)DELIVERED);
          hr = rs->Update();
        }

        hr = rs->Close();
      }
      
      if (pMsg) {
        delete pMsg;
        pMsg = NULL;
      }

      releaseLock();
    } catch (_com_error &e) {
      BSTR b;
      if (NULL != e.ErrorInfo()) {
        e.ErrorInfo()->GetDescription(&b);
        char szErr[256];
        sprintf(szErr, "Error: %s\n",_com_util::ConvertBSTRToString(b));
        //::MessageBox(NULL, szErr, "Message Queue Error", MB_ICONERROR | MB_OK);
        OutputDebugString(szErr);
      }
    } catch (...) {
      bTerminated = true;
    }
    ::Sleep(1000);
  }
}

const char* SQLServerAdapter::getCurrTime()
{
  HRESULT hr = S_FALSE;
  ADODB::_ConnectionPtr c; 
  hr = c.CreateInstance(__uuidof(ADODB::Connection), NULL, CLSCTX_INPROC_SERVER);  
  if (S_OK != hr) throw("CreateInstance failed");
  hr = c->Open(bstr_t(CONNECTSTRING), L"", L"", -1);  

  _bstr_t sql = " select getdate() as currtime ";

  ADODB::_RecordsetPtr rs;  
  rs.CreateInstance (__uuidof(ADODB::Recordset));  
  rs->Open(sql, c.GetInterfacePtr(), ADODB::adOpenDynamic, ADODB::adLockOptimistic, ADODB::adCmdText);  

  _bstr_t currtime;

  if (!rs->ADOEOF) hr = rs->MoveFirst();
  if (S_OK == hr && !rs->ADOEOF) {
    currtime = _bstr_t(rs->Fields->GetItem(L"currtime")->GetValue());
  } 

  try { rs->Close(); } catch (...) {}
  try { c->Close(); } catch (...) {}

  m_pszTemp = new char[currtime.length()+1];
  memset(m_pszTemp, 0, currtime.length()+1);
  strncpy(m_pszTemp, currtime, currtime.length());
  return m_pszTemp;
}

void SQLServerAdapter::dispatch(const char * msg)
{
  HRESULT hr;
  try {     
    _bstr_t szMsg = "<?xml version=\"1.0\" encoding=\"iso-8859-1\"?>"
                    "<!DOCTYPE message ["
                    "<!ELEMENT message (header, body)>"
                    "<!ELEMENT header (id?, date?, refid?, sender, status?, subject)>"
                    "<!ELEMENT id      (#PCDATA)>"
                    "<!ELEMENT date    (#PCDATA)>"
                    "<!ELEMENT sender  (#PCDATA)>"
                    "<!ELEMENT status  (#PCDATA)>"
                    "<!ELEMENT subject (#PCDATA)>"
                    "<!ELEMENT body    (#PCDATA)>"
                    "]>"
                    "<message>"
                    "<header>"
                    "<date>"+_bstr_t(getCurrTime())+"</date>"
                    "<sender>"+_bstr_t(this->getSender())+"</sender>"
                    "<subject>"+_bstr_t(this->getSubjectText())+"</subject>"
                    "</header>"
                    "<body>"+_bstr_t(msg)+"</body>"
                    "</message>";

    bstr_t sql = _bstr_t(" INSERT INTO MESSAGE_QUEUE (STATUS, SENDER, SUBJECT, MESSAGE) VALUES ('"+ _bstr_t(PENDING) +"', '"+ _bstr_t(this->szSender) +"', '"+ _bstr_t(DatabaseAdapter::SubjectText(this->subject)) +"', CONVERT(xml, '"+ szMsg +"', 2)) \n");
    //OutputDebugString(sql);
    hr = cn->Execute(sql, NULL, ADODB::adExecuteNoRecords);
  } catch (_com_error &e) {
    BSTR b;    
    if (e.ErrorInfo()) {
      e.ErrorInfo()->GetDescription(&b);
      char szErr[256];
      sprintf(szErr, "Error:%s\n",_com_util::ConvertBSTRToString(b));
      ::MessageBox(NULL, szErr, "Message Queue Error", MB_ICONERROR | MB_OK);
    }
  } catch (...) {
  }
}

void SQLServerAdapter::setStatus(E_STATUS stat, infobus::types::Message * pMsg)
{
  HRESULT hr;
  try {
    std::stringstream ss;
    ss << " UPDATE MESSAGE_QUEUE SET STATUS = '"<< stat <<"' "
       << " WHERE MESSAGE_ID = '"<< pMsg->getId() <<"' ";
    
    bstr_t sql;
    sql = _bstr_t(ss.str().c_str());
    hr = cn->Execute(sql, NULL, ADODB::adExecuteNoRecords);
  } catch (_com_error &e) {
    BSTR b;
    e.ErrorInfo()->GetDescription(&b);
    char szErr[256];
    sprintf(szErr, "Error:%s\n",_com_util::ConvertBSTRToString(b));
    ::MessageBox(NULL, szErr, "Message Queue Error", MB_ICONERROR | MB_OK);
  } catch (...) {
  }
}

void SQLServerAdapter::execute(const char* pszSql)
{
    HRESULT hr = S_FALSE;
    _bstr_t sql = _bstr_t(pszSql);
    
    hr = cn->Execute(sql, NULL, ADODB::adExecuteNoRecords);
    
    if (S_OK != hr) {
      std::stringstream ss;
      ss << "Error: MetaStorageBase::Execute, "
         << "sql["<< pszSql <<"], "
         << std::endl; 
      OutputDebugString(ss.str().c_str());
    }
}
